package ex.tableapi.window;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

public class ExGroupWindows extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExGroupWindows demo = new ExGroupWindows();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");

        Table result = demo.exGroupWindows(orders);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }

    private Table exGroupWindows(Table orders) {
        Table groupByWindowDistinctResult = orders
                .window(Tumble
                        .over(lit(1).hour())
                        .on($("rowtime"))
                        .as("w")
                )
                .groupBy($("w"), $("a"))
//                .select($("a"),$("b").sum());
                //eg:id:1的数据
                // 当前时间 2024-01-26 19:05:07
                // 开始    2024-01-26T19:00
                // 结束    2024-01-26T20:00
                .select($("a"), $("w").start(), $("w").end(), $("w").rowtime(), $("b").count());
        // aggregate and add window start, end, and rowtime timestamps

//        1> +I[liji, 1100]
//        1> +I[liji, 2400]
//        1> +I[zy, 1600]
//        1> +I[zy, 1500]
//        1> +I[zy, 650]

        return groupByWindowDistinctResult;
    }


}
